#include "skynet.h"
#include "skynet_socket.h"
#include "databuffer.h"
#include "hashid.h"

#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <stdint.h>
#include <stdio.h>
#include <stdarg.h>

#define BACKLOG 128

#define MESSAGE_CGATE_TYPE 101
#define BOYAA_HEADER_LEN 8
#define BOYAA_MAX_PACKAGE_SIZE 1024 * 16

struct connection {
	int id;	// skynet_socket id
	uint32_t agent;
	uint32_t client;
	char remote_name[32];
	struct databuffer buffer;
};

struct service_ipksocket_gate {
	struct skynet_context *ctx;
	int listen_id;
	uint32_t watchdog;
	uint32_t broker;
	int client_tag;
	int header_size;
	int max_connection;
	struct hashid hash;
	struct connection *conn;
	// todo: save message pool ptr for release
	struct messagepool mp;
};

struct service_ipksocket_gate *
service_ipksocket_gate_create(void) {
	struct service_ipksocket_gate * g = skynet_malloc(sizeof(*g));
	memset(g,0,sizeof(*g));
	g->listen_id = -1;
	return g;
}

void
service_ipksocket_gate_release(struct service_ipksocket_gate *g) {
	int i;
	struct skynet_context *ctx = g->ctx;
	for (i=0;i<g->max_connection;i++) {
		struct connection *c = &g->conn[i];
		if (c->id >=0) {
			skynet_socket_close(ctx, c->id);
		}
	}
	if (g->listen_id >= 0) {
		skynet_socket_close(ctx, g->listen_id);
	}
	messagepool_free(&g->mp);
	hashid_clear(&g->hash);
	skynet_free(g->conn);
	skynet_free(g);
}

static void
_forward_agent(struct service_ipksocket_gate * g, int fd, uint32_t agentaddr, uint32_t clientaddr) {
	int id = hashid_lookup(&g->hash, fd);
	if (id >=0) {
		struct connection * agent = &g->conn[id];
		agent->agent = agentaddr;
		agent->client = fd;
		agent->id = id;
		//skynet_error(g->ctx, "connect get_fd_ok! _forward_agent fd=%d, hash_id=%d,  agent=%d, client=%d", fd, id, agentaddr, clientaddr);
	}
}

static void
_report(struct service_ipksocket_gate * g, const char * data, ...) {
	if (g->watchdog == 0) {
		return;
	}
	struct skynet_context * ctx = g->ctx;
	va_list ap;
	va_start(ap, data);
	char tmp[1024];
	int n = vsnprintf(tmp, sizeof(tmp), data, ap);
	va_end(ap);

	//skynet_error(g->ctx, "_report watchdog=%d, tmp=%s, sz=%d ", g->watchdog, tmp, n);

	skynet_send(ctx, 0, g->watchdog, MESSAGE_CGATE_TYPE,  0, tmp, n);
}

static void
_forward_boyaa(struct service_ipksocket_gate *g, struct connection * c, int size) {
	struct skynet_context * ctx = g->ctx;
	int fd = c->id;
	if (fd <= 0) {
		//socket error
		skynet_error(g->ctx, "_forward_boyaa socket error");
		return;
	}

	void *buffer = skynet_malloc(size + BOYAA_HEADER_LEN);
    memcpy(buffer, c->buffer.boyaa_header, BOYAA_HEADER_LEN);

    databuffer_read(&c->buffer, &g->mp, buffer + BOYAA_HEADER_LEN, size);

    //skynet_error(g->ctx, "forward package to boyaa robot agent, %s", (char*)buffer);

	skynet_send(ctx, c->client, c->agent, g->client_tag | PTYPE_TAG_DONTCOPY, 0, buffer, size + BOYAA_HEADER_LEN);
    
	return;
}

static void
dispatch_message(struct service_ipksocket_gate *g, struct connection *c, int id, void * data, int sz) {
	databuffer_push(&c->buffer,&g->mp, data, sz);
	for (;;) {
		int size = databuffer_readheader_boyaa(&c->buffer, &g->mp, BOYAA_HEADER_LEN);
		if (c->buffer.header < 0 || c->buffer.header > BOYAA_MAX_PACKAGE_SIZE) {
            struct skynet_context *ctx = g->ctx;
            databuffer_clear(&c->buffer, &g->mp);
            skynet_socket_close(ctx, id);
            skynet_error(ctx, "wrong socket msg size:%d", c->buffer.header);
            return;
        }

		if (size < 0) {
			return;
		} else if (size >= 0) {
			if (size >= 0x1000000) {
				struct skynet_context * ctx = g->ctx;
				databuffer_clear(&c->buffer,&g->mp);
				skynet_socket_close(ctx, id);
				skynet_error(ctx, "Recv socket message > 16M");
				return;
			} else {
				_forward_boyaa(g, c, size);
				databuffer_reset(&c->buffer);
			}
		}
	}
}

static void
dispatch_socket_message(struct service_ipksocket_gate *g, const struct skynet_socket_message * message, int sz) {
	struct skynet_context * ctx = g->ctx;

	switch(message->type) {
	case SKYNET_SOCKET_TYPE_DATA: {
		int id = hashid_lookup(&g->hash, message->id);
		if (id>=0) {
			struct connection *c = &g->conn[id];
			dispatch_message(g, c, message->id, message->buffer, message->ud);
		} else {
			skynet_error(ctx, "Drop unknown connection %d message", message->id);
			skynet_socket_close(ctx, message->id);
			skynet_free(message->buffer);
		}
		break;
	}
	case SKYNET_SOCKET_TYPE_CONNECT: {
		int id = hashid_lookup(&g->hash, message->id);
		if (id<0) {
			skynet_error(ctx, "Close unknown connection %d", message->id);
			skynet_socket_close(ctx, message->id);
		}
		else {
			struct connection *c = &g->conn[id];
			//skynet_error(ctx, "agent:%d  connected! fd=%d", c->agent, message->id);

			_report(g, "connected#%d#%d", c->agent, message->id);
		}
		break;
	}
	case SKYNET_SOCKET_TYPE_CLOSE:
	case SKYNET_SOCKET_TYPE_ERROR: {
		int id = hashid_remove(&g->hash, message->id);
		if (id>=0) {
			struct connection *c = &g->conn[id];
			databuffer_clear(&c->buffer,&g->mp);
			memset(c, 0, sizeof(*c));
			c->id = -1;
			_report(g, "close#%d", message->id);
		}
		break;
	}
	case SKYNET_SOCKET_TYPE_WARNING:
		skynet_error(ctx, "fd (%d) send buffer (%d)K", message->id, message->ud);
		_report(g, "warning#%d", message->id);
		break;
	}
}

static const char* 
_get_lua_str(struct skynet_context * ctx, const char* msg, int sz)
{
	char *pchTmp = NULL;
	char chBuffer[sz+1];
	memcpy(chBuffer, msg, sz);
	chBuffer[sz] = '\0';

	pchTmp = chBuffer;
	char *pchDilem = "#";
	char *pchStrTmpIn = NULL;

	pchTmp = strtok_r( pchTmp, pchDilem, &pchStrTmpIn);

	while(NULL != ( pchTmp = strtok_r( pchTmp, pchDilem, &pchStrTmpIn) )) {
		pchTmp = NULL;
	}

	return NULL;
}

static void
_ctrl_lua(struct service_ipksocket_gate * g, const void * msg, int sz, uint32_t source) {
	struct skynet_context * ctx = g->ctx;
	char tmp[sz+1];
	memcpy(tmp, msg, sz);
	tmp[sz] = '\0';
	char * command = tmp;
	int i;
	if (sz == 0)
		return;
	for (i=0;i<sz;i++) {
		if (command[i]==' ') {
			break;
		}
	}

	char *pchStrTmpIn = NULL;
	command = strtok_r( command, "#", &pchStrTmpIn);
	if (strcmp(command, "initialization")==0) {
		const char* watchdog_str = strtok_r( NULL, "#", &pchStrTmpIn);
		g->watchdog = strtol(watchdog_str , NULL, 10);
		skynet_error(ctx, "###############[Init Finish] (CGATE)g->watchdog=%d###############", g->watchdog);
		return ;
	}
	if (strcmp(command, "connect")==0) {
		const char* ip = strtok_r( NULL, "#", &pchStrTmpIn);

		const char* port_str = strtok_r( NULL, "#", &pchStrTmpIn);
		int port = strtol(port_str , NULL, 10);

		const char* agent_str = strtok_r( NULL, "#", &pchStrTmpIn);
		uint32_t agent = strtol(agent_str , NULL, 10);

		if (hashid_full(&g->hash)) {
            _report(g, "error#%d", source, 0);
            return;
        }
		else {
			int fd = skynet_socket_connect(ctx, ip, port);
			if (fd > 0) {
				hashid_insert(&g->hash, fd);
	 			_forward_agent(g, fd, agent, 0);
			}
		}
		return;
	}
	if (strcmp(command, "kick")==0) {
		const char* fd_str = strtok_r( NULL, "#", &pchStrTmpIn);
		int the_fd = strtol(fd_str , NULL, 10);
		int id = hashid_lookup(&g->hash, the_fd);
		if (id >= 0) {
            skynet_socket_close(ctx, the_fd);
			//skynet_error(ctx, "cgate kick success! fd=%d", the_fd);
        }else {
			skynet_error(ctx, "cgate kick error! fd=%d", the_fd);
		}
	
		return ;
	}
	if (strcmp(command, "debug_infos")==0) {
		int count = 0;
		for (i=0;i<g->max_connection;i++) {
			struct connection *c = &g->conn[i];
			if (c->id >=0) {
				++count;
				skynet_error(ctx, "c->id=%d fd=%u agent=%u", c->id, c->client, c->agent);
			}
		}
		skynet_error(ctx, "check g->conn has [%d] connecting!", count);
		return ;
	}

	skynet_error(ctx, "[service_ipksocket_gate] Unkown command : %s", command);
}

static int _cb(struct skynet_context * ctx, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
	struct service_ipksocket_gate *g = ud;

	switch(type) {
	case MESSAGE_CGATE_TYPE:
		_ctrl_lua(g , msg , (int)sz, source);
		break;
	case PTYPE_SOCKET:
		// recv socket message from skynet_socket
		dispatch_socket_message(g, msg, (int)(sz-sizeof(struct skynet_socket_message)));
		break;
	}
	
	return 0;
}

int
service_ipksocket_gate_init(struct service_ipksocket_gate *g , struct skynet_context * ctx, char * parm) {
	int max = 500;
	g->ctx = ctx;

	hashid_init(&g->hash, max);
	g->conn = skynet_malloc(max * sizeof(struct connection));
	memset(g->conn, 0, max *sizeof(struct connection));
	g->max_connection = max;
	int i;
	for (i=0;i<max;i++) {
		g->conn[i].id = -1;
	}

	g->client_tag = PTYPE_CLIENT;
	skynet_callback(ctx,g,_cb);

	skynet_handle_namehandle(skynet_context_handle(g->ctx), "cgate");
	
    return 0;
}

